import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 编程题二、
 * 假设机场的数据如下：
 *    1, "SFO"
 *    2, "ORD"
 *    3, "DFW"
 * 机场两两之间的航线及距离如下：
 *    1, 2,1800
 *    2, 3, 800
 *    3, 1, 1400
 * 用 GraphX 完成以下需求：
 *    求所有的顶点
 *    求所有的边
 *    求所有的triplets
 *    求顶点数
 *    求边数
 *    求机场距离大于1000的有几个，有哪些
 *    按所有机场之间的距离排序（降序），输出结果
 */

object Subject2 {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val sc = new SparkContext(conf)

    //定义顶点
    val vertexArray: Array[(VertexId, String)] = Array(
      (1L, "SFO"),
      (2L, "ORD"),
      (3L, "DFW")
    )
    val vertexRDD: RDD[(VertexId, String)] = sc.makeRDD(vertexArray)

    //定义边
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(1L, 2L, 1800),
      Edge(2L, 3L, 800),
      Edge(3L, 1L, 1400)
    )
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    //定义图
    val graph: Graph[String, Int] = Graph.apply(vertexRDD, edgeRDD)

    println("所有的顶点:")
    graph.vertices.foreach(println)

    println("所有的边：")
    graph.edges.foreach(println)

    println("所有的triplets：")
    graph.triplets.foreach(println)

    println(s"顶点数：${graph.vertices.count()}")
    println(s"边数：${graph.edges.count()}")

    println("机场距离大于1000的有哪几个：")
    graph.edges.filter(_.attr > 1000).foreach(println)

    println("按所有机场之间的距离排序（降序），输出结果：")
    graph.edges.sortBy(_.attr, false).collect.foreach(println)

    sc.stop()
  }
}
